package com.rengwuxian.coursecoroutines._4_flow

import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.launch
import kotlinx.coroutines.newFixedThreadPoolContext
import kotlinx.coroutines.plus
import kotlinx.coroutines.runBlocking
import kotlin.coroutines.EmptyCoroutineContext

/**
 * flowOn()操作符
 * 切线程、flowOn()、withContext()、launchIn()、channelFlow()
 *
 * flowOn()用来定制flow运行的CoroutineContext的，大多数时候，就是用它来切线程的。
 *         作用跟catch()有点类似，只管自己的上游Flow。
 *         下游的collect中的CoroutineContext是不受影响的。
 *         多个flowOn()的作用：各管各的上游，和catch()作用一样。
 * 最下游的flow，即collect部分是由启动它的协程来管理的。使用它的CoroutineContext。
 * 或者把collect代码块放在onEach()里后使用flowOn()，再调用collect().(官方推荐写法) --》纯粹使用flowOn
 * flow1.map {
 *       it + 1
 *     }.onEach {
 *       println("Data: $it - ${currentCoroutineContext()}")
 *     }.flowOn(Dispatchers.IO)
 *       .collect {}
 * 或者脱离外部协程，也可以launchIn()操作符
 * flow1.map {
 *     it + 1
 *   }.onEach {
 *     println("Data: $it - ${currentCoroutineContext()}")
 *   }.launchIn(scope + Dispatchers.IO)
 * 上面只是写法上有差异，性能上没有任何变化。看着优雅。
 *
 * withContext()也可以切换CoroutineContext，但是不能包住emit，需要更细粒度的控制。
 *
 * 连续调用两个flowOn()会导致融合特点：
 * 第一个flowOn()会创建一个新的Flow，后面的flowOn()会基于前面的Flow做修改，不会创建新的Flow。
 * flowOn(Dispatchers.IO).flowOn(Dispatchers.Default)
 * CoroutineContext会相加，融合到左边这个里面，（从右往左加）--》只会留下Dispatchers.IO
 *
 * flowOn()的底层原理就是channelFlow{},在channelFlow{}的后面使用flowOn()，也会产生融合效果。
 */
fun main() = runBlocking<Unit> {
  val scope = CoroutineScope(EmptyCoroutineContext)
  val flow1 = flow {
    println("CoroutineContext in flow(): ${currentCoroutineContext()}")
    for (i in 1..5) {
      emit(i)
    }
  }.map {
    println("CoroutineContext in map() 1: ${currentCoroutineContext()}")
    it * 2
  }.flowOn(Dispatchers.IO).flowOn(Dispatchers.Default)
    .map {
      println("CoroutineContext in map() 2: ${currentCoroutineContext()}")
      it * 2
    }.flowOn(newFixedThreadPoolContext(2, "TestPool"))
  val flow2 = channelFlow {
    println("CoroutineContext in channelFlow(): ${currentCoroutineContext()}")
    for (i in 1..5) {
      send(i)
    }
  }.map { it }.flowOn(Dispatchers.IO)
  scope.launch {
    /*flow1.map {
      it + 1
    }.onEach {
      println("Data: $it - ${currentCoroutineContext()}")
    }.flowOn(Dispatchers.IO)
      .collect {}*/
    flow2.collect()
  }
  /*flow1.map {
    it + 1
  }.onEach {
    println("Data: $it - ${currentCoroutineContext()}")
  }.launchIn(scope + Dispatchers.IO)*/
  delay(10000)
}